package com.atguigu.Util;

/**
 * @author hjy
 * @create 2023/3/14 20:36
 */

import com.atguigu.bean.TransientSink;
import com.atguigu.common.GmallConfig;
import lombok.SneakyThrows;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * 这个工具类用于将流中的数据写入到clickhouse中
 * 根据addsink()中需要的一个SinkFunction 所以我们这个类的方法应该返回一个一个SinkFunction
 */
public class ClickHouseUtil {
    //sql:insert into db.tn values(?,?,?,?,?)
    public static <T> SinkFunction<T> getSinkFunction(String sql) {
        return JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @SneakyThrows
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        /**
                         * 通过反射获取JavaBean的所有字段
                         * 如果存在一个字段是javaBean有但是clickHouse的表中没有的
                         * 需要给javaBean的这个字段加上标签（注解）
                         * 遍历字段，这里判断是否有这个注解，有则跳过这个字段
                         * 通过preparedStatement写入
                         */
                        Class<?> aClass = t.getClass();
                        Field[] fields = aClass.getDeclaredFields();
                        int offset=0;
                        for (int i = 0; i < fields.length; i++) {
                            //拿到字段
                            Field field = fields[i];
                            //判断是否有特殊注解标记，若有则跳过 这里要写一个注解类
                            TransientSink annotation = field.getAnnotation(TransientSink.class);
                            if (annotation!=null){
                                offset++;
                                continue;
                            }
                            //这里会有非法访问异常，我们直接设置它的权限，然后抛出这个异常
                            field.setAccessible(true);
                            Object value = field.get(t);
                            preparedStatement.setObject(i-offset+1,value);
                        }

                    }
                },
                JdbcExecutionOptions.builder()
                        //每5条发送一次
                        .withBatchSize(5)
                        //每一秒发送一次
                        .withBatchIntervalMs(1000L)
                        .build(),
                new JdbcConnectionOptions.
                        JdbcConnectionOptionsBuilder()
                        .withDriverName(GmallConfig.CLICKHOUSE_DRIVER)
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build()


        );
    }
}
